/*
 * Copyright (c) 2024 Black Sesame Technologies
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#ifndef _DAL_FASTDDS_SERVICE_RESP_WRITER_LISTENER_IMPL_HPP_
#define _DAL_FASTDDS_SERVICE_RESP_WRITER_LISTENER_IMPL_HPP_

#include <unordered_set>
#include <unordered_map>
#include <condition_variable>
#include "dal/client.h"
#include "dal_fastdds/utils/guid_utils.hpp"
#include <fastdds/dds/publisher/Publisher.hpp>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fastdds/dds/publisher/DataWriterListener.hpp>

using namespace eprosima::fastdds::dds;
using namespace eprosima::fastrtps::rtps;
using namespace dal_fastdds;

class DalServiceImpl;

enum class ClientPresentStatus
{
    FAILURE,  // an error occurred when checking
    MAYBE,    // reader not matched, writer still present
    YES,      // reader matched
    GONE      // neither reader nor writer
};

class ServiceRespWriterListenerImpl : public DataWriterListener
{
public:
    explicit ServiceRespWriterListenerImpl(DalServiceImpl* dal_service_impl)
        : dal_service_impl_(dal_service_impl)
    {
    }

    void on_publication_matched(DataWriter*, const PublicationMatchedStatus& info)
    {
        std::lock_guard<std::mutex> lock(mutex_);

        auto guid = iHandle2GUID(info.last_subscription_handle);
        if (info.current_count_change == 1) {
            matched_readers_set_.insert(guid);
        } else if (info.current_count_change == -1) {
            matched_readers_set_.erase(guid);

            auto endpoint = clients_endpoints_map_.find(guid);
            if (endpoint != clients_endpoints_map_.end()) {
                clients_endpoints_map_.erase(endpoint->second);
                clients_endpoints_map_.erase(guid);
            }
        } else {
            return;
        }

        cv_.notify_all();
    }

    template <class Rep, class Period>
    bool wait_for_subscription(const GUID_t& guid,
                               const std::chrono::duration<Rep, Period>& rel_time)
    {
        auto guid_is_present = [this, guid]() -> bool {
            return matched_readers_set_.find(guid) != matched_readers_set_.end();
        };

        std::unique_lock<std::mutex> lock(mutex_);
        return cv_.wait_for(lock, rel_time, guid_is_present);
    }

    template <class Rep, class Period>
    ClientPresentStatus
    check_for_subscription(const GUID_t& guid,
                           const std::chrono::duration<Rep, Period>& max_blocking_time)
    {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            // Check if the guid is still in the map
            if (clients_endpoints_map_.find(guid) == clients_endpoints_map_.end()) {
                // Client is gone
                return ClientPresentStatus::GONE;
            }
        }

        // Wait for subscription
        if (!wait_for_subscription(guid, max_blocking_time)) {
            return ClientPresentStatus::MAYBE;
        }
        return ClientPresentStatus::YES;
    }

    void endpoint_erase_if_exists(const GUID_t& endpointGuid)
    {
        std::lock_guard<std::mutex> lock(mutex_);
        auto endpoint = clients_endpoints_map_.find(endpointGuid);
        if (endpoint != clients_endpoints_map_.end()) {
            clients_endpoints_map_.erase(endpoint->second);
            clients_endpoints_map_.erase(endpointGuid);
        }
    }

    void endpoint_add_reader_and_writer(const GUID_t& readerGuid, const GUID_t& writerGuid)
    {
        std::lock_guard<std::mutex> lock(mutex_);
        clients_endpoints_map_.emplace(readerGuid, writerGuid);
        clients_endpoints_map_.emplace(writerGuid, readerGuid);
    }

private:
    DalServiceImpl* dal_service_impl_;
    std::unordered_set<GUID_t, hash_fastrtps_guid> matched_readers_set_;
    std::unordered_map<GUID_t, GUID_t, hash_fastrtps_guid> clients_endpoints_map_;
    std::mutex mutex_;
    std::condition_variable cv_;
};

#endif  // _DAL_FASTDDS_SERVICE_RESP_WRITER_LISTENER_IMPL_HPP_